package demo.db.tsdb.influxDB;

import demo.vo.tsdb.DataPoint;
import org.influxdb.BatchOptions;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.Point;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;

public class InfluxDbDemo {

    private static final Logger logger = LoggerFactory.getLogger(InfluxDbDemo.class);
    public static String host = "http://192.168.60.89:8428";
    private InfluxDB client;

    public static InfluxDB newInfluxDB() {
        BatchOptions batchOptions = BatchOptions.DEFAULTS
                // 设置在自动刷新批量数据到 InfluxDB 之前要等待的时间,如果缓冲区中积累的数据点数未满，客户端将等待多久（不论缓冲区的大小）强制将数据写入数据库。
                .flushDuration(1)
                // 在批量写入时，客户端在进行一次 "flush" 操作前可以容纳的最大动作数（数据点数）
                .actions(300)
                // 客户端为失败的写操作维护一个缓冲区，以便稍后重试写操作。这可能有助于克服临时网络问题或InfluxDB负载峰值。
                // 当缓冲区已满并写入新的点时，缓冲区中最老的条目将丢失。要禁用此功能，请将缓冲区限制设置为小于getActions的值
                .bufferLimit(50000)
                .exceptionHandler(new BiConsumer<Iterable<Point>, Throwable>() {
                    @Override
                    public void accept(Iterable<Point> points, Throwable throwable) {
                        logger.error("Batch write error", throwable);
                    }
                });
        return InfluxDBFactory.connect(host, "admin", "admin")
                .enableBatch(batchOptions)
                .setDatabase("");
    }

    public static void addPoint(String measurement, Map<String, String> tag, DataPoint data) {
        Point point = Point
                .measurement(measurement)
                .tag(tag)
                .time(data.getTime() / 1000, TimeUnit.SECONDS)
                .addField("total", data.getValue())
                .build();
        System.out.println("Point : " + point);
        BatchOptions batchOptions = BatchOptions.DEFAULTS.flushDuration(1)
                .actions(300)
                .bufferLimit(50000)
                .exceptionHandler((points, throwable) -> {
                    throwable.printStackTrace();
                });
        InfluxDB influxDB = InfluxDBFactory
                .connect(host, "admin", "admin")
                .enableBatch(batchOptions);

        String lineProtocol = point.lineProtocol();
        System.out.println(lineProtocol);
        influxDB.write("my", "autodel_3d", InfluxDB.ConsistencyLevel.ONE, lineProtocol);
    }

    @Before
    public void setUp() throws Exception {
        client = newInfluxDB();
    }

    @After
    public void tearDown() throws Exception {
        if (client != null) {
            client.close();
        }
    }

    public void writeArenaPoints(List<Point> points) {

        try {
            client.write("xdcsMetric",
                    "autodel_3d",
                    InfluxDB.ConsistencyLevel.ONE,
                    points.stream().map(Point::lineProtocol).collect(Collectors.toList()));
        } catch (Throwable e) {
            logger.error("write direct to influxdb error! influxDB.version = {}, error msg = {}",
                    client.version(),
                    e.getMessage());
        }

    }

    @Test
    public void testWrite() {
        String measurement = "test_vm_request";
        long now = System.currentTimeMillis();
        Map<String, String> tags = new HashMap<>();
        tags.put("app", "demo-app");
        tags.put("code", "501");
        Point point = Point.measurement(measurement) // 自定义测量名称
                .time(now, TimeUnit.MILLISECONDS) // 使用当前时间戳
                .tag(tags) // 添加标签
                .addField("total", 23.5) // 添加字段数据
                .build();
        System.out.println(point);
        System.out.println(point.lineProtocol());
        client.write(point);
    }

    @Test
    public void testBatchWrite() {
        List<Point> points = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            points.add(Point.measurement("test_vm_request")
                    .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS)
                    .tag("app", "demo-app")
                    .tag("code", "50" + i)
                    .addField("total", 1)
                    .build());
        }
        // test_vm_request,app=demo-app,code=500 total=1i 1734329487849000000
        List<String> records = points.stream().map(Point::lineProtocol).collect(Collectors.toList());
        System.out.println(records);
        client.write(records);
    }
}
